/******************************************************************************/
/* @ All Rights Reserved Copyright (C) 2014 broada,Ltd.                       */
/*                                                                            */
/*  NAME      = kafka_response.c                                              */
/*  NOTE      = kafka响应解析                                                 */
/*  DATE      = 2015/03/19 by zhengxd                                         */
/******************************************************************************/
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <stdint.h>

#include "kafka_core.h"
#include "kafka_comm.h"
#include "kafka_response.h"

/*******************************************************************************
*  FUNC     :  追加fetch的上报信息
*  ARGS     :  p_st_sess_body (上报list)
*           :  p_st_private(私有信息)
*  RTN      :
*  NOTE     :
*******************************************************************************/
static int __kafka_add_fetch_resp(Frreport_body_data *p_st_sess_body
                                  , L7session_kafka_private *p_st_private)
{
    int i_ret = FR_SUCC;
    int i_loop = 0;
    unsigned int ui_type = 0;
    unsigned int ui_text = 0;
    L7_resp_kafka *p_st_resp = p_st_private->p_st_resp;
    char arr_type[128] = {0};
    char arr_text[128] = {0};
    snprintf(arr_type, 128, KAFKA_REPORT_KEY_RESP_TYPE);
    snprintf(arr_text, 128, KAFKA_REPORT_KEY_RESP_TEXT);
    ui_type -= 128 - strlen(KAFKA_REPORT_KEY_RESP_TYPE);
    ui_text -= 128 - strlen(KAFKA_REPORT_KEY_RESP_TEXT);
    char *p_type = arr_type;
    char *p_text = arr_text;
    p_type += strlen(KAFKA_REPORT_KEY_RESP_TYPE);
    p_text += strlen(KAFKA_REPORT_KEY_RESP_TEXT);
    Frlist_node *p_st_node_topic = NULL;
    Frlist_node *p_st_node_partition = NULL;
    Frlist_node *p_st_node_message_set = NULL;
    Kafka_topic_info *p_st_topic_req = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Kafka_message_set_info *p_st_message_set = NULL;

    p_st_node_topic = FRLIST_FIRST_NODE(&p_st_resp->st_topics);
    while (p_st_node_topic)
    {
        p_st_topic_req = (Kafka_topic_info *)FRLIST_GET_NODE_DATA(p_st_node_topic);
        p_st_node_partition = FRLIST_FIRST_NODE(&p_st_topic_req->st_partitions);
        while (p_st_node_partition)
        {
            p_st_partition = (Kafka_partition_info *)FRLIST_GET_NODE_DATA(p_st_node_partition);
            p_st_node_message_set = FRLIST_FIRST_NODE(&p_st_partition->st_messages);
            while (p_st_node_message_set)
            {
                p_st_message_set = (Kafka_message_set_info *)FRLIST_GET_NODE_DATA(p_st_node_message_set);
                i_loop++;
                snprintf(p_type, ui_type, "%d", i_loop);
                snprintf(p_text, ui_text, "%d", i_loop);
                /* add type */
                i_ret = frreport_body_l7_add_str(p_st_sess_body, arr_type, "text");
                if (i_ret != FR_SUCC)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                /* add text */
                i_ret = frreport_body_l7_add_str(p_st_sess_body, arr_text
                                                , p_st_message_set->p_value ? p_st_message_set->p_value : "");
                if (i_ret != FR_SUCC)
                {
                    return APM_L7_ANALYZE_FAIL;
                }
                p_st_node_message_set = FRLIST_NEXT_NODE(p_st_node_message_set);
            }
            p_st_node_partition = FRLIST_NEXT_NODE(p_st_node_partition);
        }
        p_st_node_topic = FRLIST_NEXT_NODE(p_st_node_topic);
    }

    i_ret = frreport_body_l7_add_int(p_st_sess_body, KAFKA_REPORT_KEY_RESP_SIZE, i_loop);
    if (i_ret != FR_SUCC)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     : 上报response
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*  RTN      :
*  NOTE     :
*******************************************************************************/
static int __kafka_report_resp(Conntrack *p_st_conn
                               , Frhash_header *p_st_hash_report
                               , L7session_kafka_private *p_st_private)
{
    int i_ret = FR_SUCC;
    unsigned long ul_server_time_us = 0;
    unsigned long ul_download_time_us = 0;
    Frreport_data *p_st_report_data = NULL;
    Frreport_body_data *p_st_sess_body = NULL;
    L7_req_kafka *p_st_req = p_st_private->p_st_req;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;
    p_st_req->ul_req_sequence = frct_seq_increase(p_st_conn);
    Frlist_node *p_st_node_topic = NULL;
    Frlist_node *p_st_node_partition = NULL;
    Frlist_node *p_st_node_message_set = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Kafka_message_set_info *p_st_message_set = NULL;
    char arr_title[128] = {0};

    switch (p_st_req->i_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            p_st_node_topic = FRLIST_FIRST_NODE(&p_st_req->st_topics);
            while (p_st_node_topic)
            {
                if (FRLIST_NEXT_NODE(p_st_node_topic) == NULL)
                {
                    p_st_topic = (Kafka_topic_info *)FRLIST_GET_NODE_DATA(p_st_node_topic);
                    p_st_node_partition = FRLIST_FIRST_NODE(&p_st_topic->st_partitions);

                    while (p_st_node_partition)
                    {
                        if (FRLIST_NEXT_NODE(p_st_node_partition) == NULL)
                        {
                            p_st_partition = (Kafka_partition_info *)FRLIST_GET_NODE_DATA(p_st_node_partition);
                            p_st_node_message_set = FRLIST_FIRST_NODE(&p_st_partition->st_messages);

                            while (p_st_node_message_set)
                            {
                                if (FRLIST_NEXT_NODE(p_st_node_message_set) == NULL)
                                {
                                    p_st_message_set = (Kafka_message_set_info *)FRLIST_GET_NODE_DATA(p_st_node_message_set);
                                    /* 创建report body */
                                    p_st_sess_body = frreport_body_new_l7(p_st_conn);
                                    if (p_st_sess_body == NULL)
                                    {
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* 追加connection相关信息 */
                                    i_ret = frreport_body_l7_add_connInfo(p_st_sess_body, p_st_conn);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add session topic */
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                    , KAFKA_REPORT_KEY_SESSION_TOPIC
                                                                    , p_st_topic->p_topic_name ? p_st_topic->p_topic_name : "");
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }

                                    /* add sequence id */
                                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                                      , APM_REPORT_KEY_REQ_ID
                                                                      , p_st_req->ul_req_sequence);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add isResponseRequired */
                                    i_ret = frreport_body_l7_add_int(p_st_sess_body
                                                                    , APM_REPORT_KEY_REQ_ISRESPREQUIRED, 1);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add request start time */
                                    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                                        , APM_REPORT_KEY_REQ_START_TIMESTAMP
                                                                        , &p_st_req->st_start_timestamp);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add request end time */
                                    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                                        , APM_REPORT_KEY_REQ_END_TIMESTAMP
                                                                        , &p_st_req->st_end_timestamp);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add send time */
                                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                                      , APM_REPORT_KEY_REQ_SEND_TIME_US
                                                                      , p_st_req->ul_send_time_us);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    /* add send data bytes */
                                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                                      , APM_REPORT_KEY_REQ_DATA_BYTES
                                                                      , p_st_req->ul_request_bytes);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    if (p_st_topic->p_topic_name)
                                    {
                                        snprintf(arr_title, sizeof(arr_title), "PUT %s", p_st_topic->p_topic_name);
                                    }
                                    else
                                    {
                                        snprintf(arr_title, sizeof(arr_title), "PUT");
                                    }
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                     , APM_REPORT_KEY_TRANS_TITLE
                                                                     , arr_title);
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }

                                    /* add request topic */
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                    , KAFKA_REPORT_KEY_REQ_TOPIC
                                                                    , p_st_topic->p_topic_name ? p_st_topic->p_topic_name : "");
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }

                                    /* add request method */
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                    , KAFKA_REPORT_KEY_REQ_METHOD
                                                                    , "PUT");
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }

                                    /* add request type */
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                    , KAFKA_REPORT_KEY_REQ_TYPE
                                                                    , "text");
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }

                                    /* add request message */
                                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                                    , KAFKA_REPORT_KEY_REQ_TEXT
                                                                    , p_st_message_set->p_value ? p_st_message_set->p_value : "");
                                    if (i_ret != FR_SUCC)
                                    {
                                        frreport_body_free(&p_st_sess_body);
                                        return APM_L7_ANALYZE_FAIL;
                                    }
                                    break;
                                }
                                else
                                {
                                    p_st_node_message_set = FRLIST_NEXT_NODE(p_st_node_message_set);
                                }
                            }
                            break;
                        }
                        else
                        {
                            p_st_node_partition = FRLIST_NEXT_NODE(p_st_node_partition);
                        }
                    }
                    break;
                }
                else
                {
                    p_st_node_topic = FRLIST_NEXT_NODE(p_st_node_topic);
                }
            }
            break;

        case KAFKA_REQ_TP_FETCH:
            p_st_node_topic = FRLIST_FIRST_NODE(&p_st_req->st_topics);
            while (p_st_node_topic)
            {
                if (FRLIST_NEXT_NODE(p_st_node_topic) == NULL)
                {
                    p_st_topic = (Kafka_topic_info *)FRLIST_GET_NODE_DATA(p_st_node_topic);

                    /* 创建report body */
                    p_st_sess_body = frreport_body_new_l7(p_st_conn);
                    if (p_st_sess_body == NULL)
                    {
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* 追加connection相关信息 */
                    i_ret = frreport_body_l7_add_connInfo(p_st_sess_body, p_st_conn);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add session topic */
                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                    , KAFKA_REPORT_KEY_SESSION_TOPIC
                                                    , p_st_topic->p_topic_name ? p_st_topic->p_topic_name : "");
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }

                    /* add sequence id */
                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                      , APM_REPORT_KEY_REQ_ID
                                                      , p_st_req->ul_req_sequence);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add isResponseRequired */
                    i_ret = frreport_body_l7_add_int(p_st_sess_body
                                                    , APM_REPORT_KEY_REQ_ISRESPREQUIRED, 1);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add request start time */
                    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                        , APM_REPORT_KEY_REQ_START_TIMESTAMP
                                                        , &p_st_req->st_start_timestamp);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add request end time */
                    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                                        , APM_REPORT_KEY_REQ_END_TIMESTAMP
                                                        , &p_st_req->st_end_timestamp);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add send time */
                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                      , APM_REPORT_KEY_REQ_SEND_TIME_US
                                                      , p_st_req->ul_send_time_us);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    /* add send data bytes */
                    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                                      , APM_REPORT_KEY_REQ_DATA_BYTES
                                                      , p_st_req->ul_request_bytes);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    if (p_st_topic->p_topic_name)
                    {
                        snprintf(arr_title, sizeof(arr_title), "GET %s", p_st_topic->p_topic_name);
                    }
                    else
                    {
                        snprintf(arr_title, sizeof(arr_title), "GET");
                    }
                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                     , APM_REPORT_KEY_TRANS_TITLE
                                                     , arr_title);
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }

                    /* add request method */
                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                    , KAFKA_REPORT_KEY_REQ_METHOD
                                                    , "GET");
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }

                    /* add request topic */
                    i_ret = frreport_body_l7_add_str(p_st_sess_body
                                                    , KAFKA_REPORT_KEY_REQ_TOPIC
                                                    , p_st_topic->p_topic_name ? p_st_topic->p_topic_name : "");
                    if (i_ret != FR_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }

                    /* 追加fetch信息 */
                    i_ret = __kafka_add_fetch_resp(p_st_sess_body, p_st_private);
                    if (i_ret != APM_L7_ANALYZE_SUCC)
                    {
                        frreport_body_free(&p_st_sess_body);
                        return APM_L7_ANALYZE_FAIL;
                    }
                    break;
                }
                p_st_node_topic = FRLIST_NEXT_NODE(p_st_node_topic);
            }
            break;

        default:
            return APM_L7_ANALYZE_SUCC;
    }

    /* add response start time */
    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                        , APM_REPORT_KEY_RESP_START_TIMESTAMP
                                        , &p_st_fragment->st_start_timestamp);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }
    /* add response end time */
    i_ret = frreport_body_l7_add_timeval(p_st_sess_body
                                        , APM_REPORT_KEY_RESP_END_TIMESTAMP
                                        , &p_st_fragment->st_end_timestamp);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }

    ul_server_time_us = kafka_get_interval_time(&p_st_req->st_end_timestamp
                                             , &p_st_fragment->st_start_timestamp);
    /* add response response_time_us */
    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                      , APM_REPORT_KEY_RESP_SERVER_TIME_US
                                      , ul_server_time_us);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }

    ul_download_time_us = kafka_get_interval_time(&p_st_fragment->st_start_timestamp
                                               , &p_st_fragment->st_end_timestamp);
    /* add response response_time_us */
    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                      , APM_REPORT_KEY_RESP_DOWNLOAD_TIME_US
                                      , ul_download_time_us);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }
    /* add response firstbyte time us */
    i_ret = frreport_body_l7_add_ulong(p_st_sess_body
                                      , APM_REPORT_KEY_RESP_DATA_BYTES
                                      , p_st_fragment->ui_total_len);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }

    if(frreport_body_l7_add_int(p_st_sess_body, APM_REPORT_KEY_RESP_CODE, 0) == FR_FAIL)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }


    /* 创建report data header */
    p_st_report_data = frreport_data_l7_new(FRREPORT_TYPE_L7DATA_RESP);
    if (p_st_report_data == NULL)
    {
        frreport_body_free(&p_st_sess_body);
        return APM_L7_ANALYZE_FAIL;
    }
    /* add body into header */
    i_ret = frreport_data_add_body(p_st_report_data, p_st_sess_body);
    if (i_ret != FR_SUCC)
    {
        frreport_body_free(&p_st_sess_body);
        frreport_data_free(&p_st_report_data);
        return APM_L7_ANALYZE_FAIL;
    }

    /* 上报request数据 */
    i_ret = frreport_data_hash(p_st_hash_report, p_st_report_data);
    if (i_ret != FR_SUCC)
    {
        frreport_data_free(&p_st_report_data);
        return APM_L7_ANALYZE_FAIL;
    }

    return APM_L7_ANALYZE_SET_RESP(APM_L7_ANALYZE_SUCC);
}

/*******************************************************************************
*  FUNC     :  解析fetch response
*  ARGS     :  p_st_private (KAFKA私有数据)
*           :  i_whole_flg(packe是否完整标识)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int __kafka_fetch_resp(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_topic_count = 0;
    int i_partition_count = 0;
    int i_messageset_size = 0;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Frlist_header st_topics;
    Frlist_header st_messages;
    memset(&st_topics, 0x0, sizeof(Frlist_header));
    memset(&st_messages, 0x0, sizeof(Frlist_header));
    L7_resp_kafka *p_st_resp = p_st_private->p_st_resp;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    /* get array count */
    i_ret = kafka_get_int(p_st_data, &i_topic_count);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }
    else if (i_topic_count < 0)
    {
        return APM_L7_ANALYZE_FAIL;
    }

    /* 解析topic */
    while (i_topic_count--)
    {
        /* skip topic name */
        i_ret = kafka_skip_string(p_st_data);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(&st_topics, kafka_topic_node_free);
            return i_ret;
        }

        /* get array count */
        i_ret = kafka_get_int(p_st_data, &i_partition_count);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            frlist_remove_all(&st_topics, kafka_topic_node_free);
            return i_ret;
        }
        else if (i_partition_count < 0)
        {
            frlist_remove_all(&st_topics, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }

        /* 创建topic */
        p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
        if (p_st_topic == NULL)
        {
            frlist_remove_all(&st_topics, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));

        /* 解析partition */
        while (i_partition_count--)
        {
            /* skip partition and ErrorCode and HighwaterMarkOffset  */
            i_ret = bytebuff_skip(p_st_data, 4 + 2 + 8);
            if (i_ret != FR_SUCC)
            {
                frdbg("skip bytes failed!\n");
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_topics, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }

            /* get messageset size */
            i_ret = kafka_get_int(p_st_data, &i_messageset_size);
            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_topics, kafka_topic_node_free);
                return i_ret;
            }
            /* parse messageset */
            memset(&st_messages, 0x0, sizeof(Frlist_header));
            i_ret = kafka_parse_message_set(p_st_data, i_messageset_size
                                           , &st_messages);
            if (i_ret != APM_L7_ANALYZE_SUCC)
            {
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_topics, kafka_topic_node_free);
                return i_ret;
            }

            /* 创建partition */
            p_st_partition = (Kafka_partition_info *)frmalloc(sizeof(Kafka_partition_info));
            if (p_st_partition == NULL)
            {
                frlist_remove_all(&st_messages, kafka_message_set_node_free);
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_topics, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
            p_st_partition->st_messages = st_messages;
            /* 创建节点 */
            p_st_node = frlist_node_create(p_st_partition);
            if (p_st_node == NULL)
            {
                kafka_partition_free(&p_st_partition);
                kafka_topic_free(&p_st_topic);
                frlist_remove_all(&st_topics, kafka_topic_node_free);
                return APM_L7_ANALYZE_FAIL;
            }
            /* 追加节点 */
            frlist_push(&(p_st_topic->st_partitions), p_st_node);
        }

        /* 创建节点 */
        p_st_node = frlist_node_create(p_st_topic);
        if (p_st_node == NULL)
        {
            kafka_topic_free(&p_st_topic);
            frlist_remove_all(&st_topics, kafka_topic_node_free);
            return APM_L7_ANALYZE_FAIL;
        }
        /* 追加节点 */
        frlist_push(&st_topics, p_st_node);
    }

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_topics, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_resp->st_topics = st_topics;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析response
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*           :  i_whole_flg(packe是否完整标识)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int kafka_parse_response(Conntrack *p_st_conn
                        , Frhash_header *p_st_hash_report
                        , L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    L7_fragment_kafka *p_st_fragment = p_st_private->p_st_fragment;
    Frbyte_Packet *p_st_data = p_st_fragment->p_st_data;

    if (p_st_private->p_st_req == NULL)
    {
        frdbg("thres is no request before!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    p_st_private->p_st_resp = (L7_resp_kafka *)frmalloc(sizeof(L7_resp_kafka));
    if (p_st_private->p_st_resp == NULL)
    {
        frdbg("create response failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_private->p_st_resp, 0x0, sizeof(L7_resp_kafka));

    /* skip CorrelationId */
    i_ret = bytebuff_skip(p_st_data, 4);
    if (i_ret != FR_SUCC)
    {
        frdbg("skip CorrelationId failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }

    switch (p_st_private->p_st_req->i_req_type)
    {
        case KAFKA_REQ_TP_PRODUCE:
            i_ret = APM_L7_ANALYZE_SUCC;
            break;

        case KAFKA_REQ_TP_FETCH:
            i_ret = __kafka_fetch_resp(p_st_private);
            if (i_ret == APM_L7_ANALYZE_FAIL)
            {
                frdbg("parse fetch response failed!\n");
                return APM_L7_ANALYZE_FAIL;
            }
            break;

        case KAFKA_REQ_TP_OFFSET:
        case KAFKA_REQ_TP_METADATA:
        case KAFKA_REQ_TP_OFFSET_COMMIT:
        case KAFKA_REQ_TP_OFFSET_FETCH:
        case KAFKA_REQ_TP_CONSUMER_METADATA:
            return APM_L7_ANALYZE_SUCC;

        default:
            return APM_L7_ANALYZE_FAIL;
    }

    i_ret = __kafka_report_resp(p_st_conn, p_st_hash_report, p_st_private);
    return i_ret;
}


/*******************************************************************************
*  FUNC     :  解析fetch response
*  ARGS     :  p_st_private (KAFKA私有数据)
*           :  i_whole_flg(packe是否完整标识)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int __kafka_fetch_resp_7(L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;
    int i_err_code = 0;
    int i_message_len = 0;
    Frlist_node *p_st_node = NULL;
    Kafka_topic_info *p_st_topic = NULL;
    Kafka_partition_info *p_st_partition = NULL;
    Frlist_header st_topics;
    Frlist_header st_messages;
    memset(&st_topics, 0x0, sizeof(Frlist_header));
    memset(&st_messages, 0x0, sizeof(Frlist_header));
    L7_resp_kafka *p_st_resp = p_st_private->p_st_resp;
    Frbyte_Packet *p_st_data = p_st_private->p_st_fragment->p_st_data;

    i_ret = kafka_get_short(p_st_data, &i_err_code);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        return i_ret;
    }

    /* 创建topic */
    p_st_topic = (Kafka_topic_info *)frmalloc(sizeof(Kafka_topic_info));
    if (p_st_topic == NULL)
    {
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_topic, 0x0, sizeof(Kafka_topic_info));
    i_message_len = bytebuff_remaining(p_st_data);
    if (i_message_len && p_st_private->i_msg_version == 0)
    {
        i_ret = kafka_get_msg_version(p_st_data, i_message_len
                                     , &p_st_private->i_msg_version);
        if (i_ret != APM_L7_ANALYZE_SUCC)
        {
            kafka_topic_free(&p_st_topic);
            return i_ret;
        }
    }
    /* parse messageset */
    memset(&st_messages, 0x0, sizeof(Frlist_header));
    i_ret = kafka_parse_messages(p_st_data, i_message_len
                                , p_st_private->i_msg_version, &st_messages);
    if (i_ret != APM_L7_ANALYZE_SUCC)
    {
        kafka_topic_free(&p_st_topic);
        return i_ret;
    }

    /* 创建partition */
    p_st_partition = (Kafka_partition_info *)frmalloc(sizeof(Kafka_partition_info));
    if (p_st_partition == NULL)
    {
        frlist_remove_all(&st_messages, kafka_message_set_node_free);
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    p_st_partition->st_messages = st_messages;
    /* 创建节点 */
    p_st_node = frlist_node_create(p_st_partition);
    if (p_st_node == NULL)
    {
        kafka_partition_free(&p_st_partition);
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* 追加节点 */
    frlist_push(&(p_st_topic->st_partitions), p_st_node);

    /* 创建节点 */
    p_st_node = frlist_node_create(p_st_topic);
    if (p_st_node == NULL)
    {
        kafka_topic_free(&p_st_topic);
        return APM_L7_ANALYZE_FAIL;
    }
    /* 追加节点 */
    frlist_push(&st_topics, p_st_node);

    /* 格式check */
    if (bytebuff_remaining(p_st_data))
    {
        frlist_remove_all(&st_topics, kafka_topic_node_free);
        return APM_L7_ANALYZE_FAIL;
    }

    p_st_resp->st_topics = st_topics;
    return APM_L7_ANALYZE_SUCC;
}

/*******************************************************************************
*  FUNC     :  解析response
*  ARGS     :  p_st_conn (入力链接信息)
*           :  p_st_hash_report(Frreport_data)
*           :  p_st_private(KAFKA私有数据)
*           :  i_whole_flg(packe是否完整标识)
*  RTN      :  参照module.h
*  NOTE     :
*******************************************************************************/
int kafka_parse_response_7(Conntrack *p_st_conn
                          , Frhash_header *p_st_hash_report
                          , L7session_kafka_private *p_st_private)
{
    int i_ret = APM_L7_ANALYZE_SUCC;

    if (p_st_private->p_st_req == NULL)
    {
        frdbg("thres is no request before!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    p_st_private->p_st_resp = (L7_resp_kafka *)frmalloc(sizeof(L7_resp_kafka));
    if (p_st_private->p_st_resp == NULL)
    {
        frdbg("create response failed!\n");
        return APM_L7_ANALYZE_FAIL;
    }
    memset(p_st_private->p_st_resp, 0x0, sizeof(L7_resp_kafka));

    switch (p_st_private->p_st_req->i_req_type)
    {
        case KAFKA_REQ_TP_FETCH:
        case KAFKA_REQ_TP_MULTIPRODUCE:
            i_ret = __kafka_fetch_resp_7(p_st_private);
            if (i_ret == APM_L7_ANALYZE_FAIL)
            {
                frdbg("parse fetch response failed!\n");
                return APM_L7_ANALYZE_FAIL;
            }
            break;

        default:
            return APM_L7_ANALYZE_SUCC;
    }

    i_ret = __kafka_report_resp(p_st_conn, p_st_hash_report, p_st_private);
    return i_ret;
}
